# -*- coding:utf-8 -*-
from pyspark import SparkConf,SparkContext
from urllib import request
import json
import shutil
from urllib import parse
district_dict=dict()

"""
下载html
"""
def download_html(url):
    req = request.Request(url)
    req.add_header('User-Agent', 'Mozilla/6.0 (iPhone; CPU iPhone OS 8_0 like Mac OS X) AppleWebKit/536.26 (KHTML, like Gecko) Version/8.0 Mobile/10A5376e Safari/8536.25')
    with request.urlopen(req) as f:
        data = f.read().decode("utf-8")
        return data
"""
地址转经纬度
"""
def address_to_location(address):
    key = address
    ress = "未知"
    if district_dict.get(key)==None:
        city = parse.quote("南宁")
        address = parse.quote(address)
        url = "https://restapi.amap.com/v3/geocode/geo?key=667dd97cc9ca502c88dee11bfdabf056&=city=%s&address=%s&output=JSON" % (city,address)
        res = download_html(url)
        print(key+res)
        data = json.loads(res)

        if data['status'] == '1':
            if len(data['geocodes'])!=0:
                a = data['geocodes'][0]['district']
                if isinstance(a,str):
                    ress = a
                    district_dict[key]=ress
    else:
        ress = district_dict.get(key)
    return ress
def tran(item):
    if item[2] == '未知':
        ress = address_to_location(item[1])
        return (item[0],item[1],ress)
    return (item[0],item[1],item[2])
if __name__ == "__main__":
    path = "D://app/学校信息20190611"

    shutil.rmtree(path)
    conf = SparkConf().setMaster("local[1]").setAppName("SparkDemo")
    sc = SparkContext(conf=conf)
    res = sc.textFile("./学校信息20190611.txt").map(lambda item: item.split("\t")).filter(lambda item:len(item)==3).map(tran)\
    .map(lambda item: "%s\t%s\t%s" % (item[0],item[1],item[2])).collect()
    sc.parallelize(res).coalesce(1).saveAsTextFile(path)